package org.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.atguigu.gmall.realtime.app.BaseApp;
import org.atguigu.gmall.realtime.bean.TrafficHomeDetailPageViewBean;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.util.AtguiguUtil;
import org.atguigu.gmall.realtime.util.FlinkSinkUtil;

import java.time.Duration;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dws
 * Author: ZengHaiFeng
 * CreateTime:2023/5/4 11:30:58
 * Description: Todo
 * Version：1.0
 */
public class Dws_03_DwsTrafficHomeDetailPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new Dws_03_DwsTrafficHomeDetailPageViewWindow()
                .init(40003,
                        2,
                        Constant.TOPIC_DWD_TRAFFIC_PAGE,
                        "DwsTrafficHomeDetailPageViewWindow");
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> ds) {
        // filter home page and detail page
        SingleOutputStreamOperator<JSONObject> filterStream = filterHomeAndGoodDetail(ds);
        // parse to pojo
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> parseStream = parseToPojo(filterStream);
        // window and aggregation
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> resultStream = windowAndAgg(parseStream);
        // output to ClickHouse
        //resultStream.print();
        writeToClickhouse(resultStream);
    }

    private void writeToClickhouse(SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getClickHouseSink(
                "dws_traffic_home_detail_page_view_window",
                TrafficHomeDetailPageViewBean.class));
    }

    private SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> windowAndAgg(SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> parseStream) {
        return parseStream
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TrafficHomeDetailPageViewBean>() {
                    @Override
                    public TrafficHomeDetailPageViewBean reduce(
                            TrafficHomeDetailPageViewBean value1,
                            TrafficHomeDetailPageViewBean value2) {
                        value1.setHomeUvCt(value1.getHomeUvCt() + value2.getHomeUvCt());
                        value1.setGoodDetailUvCt(value1.getGoodDetailUvCt() + value2.getGoodDetailUvCt());
                        return value1;
                    }
                },
                new AllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean,TimeWindow>() {
                    @Override
                    public void apply(
                            TimeWindow window,
                            Iterable<TrafficHomeDetailPageViewBean> values,
                            Collector<TrafficHomeDetailPageViewBean> out) {
                        TrafficHomeDetailPageViewBean bean = values.iterator().next();
                        bean.setStt(AtguiguUtil.tsToDateTime(window.getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(window.getEnd()));
                        bean.setTs(System.currentTimeMillis());
                        out.collect(bean);
                    }
                }
                );
    }

    private SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> parseToPojo(SingleOutputStreamOperator<JSONObject> filterStream) {
        return filterStream
                .keyBy(value -> value.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>() {
                    private ValueState<String> lastVisitHomeDateState;
                    private ValueState<String> lastVisitGoodDetailDateState;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastVisitHomeDateState = getRuntimeContext().getState(
                                new ValueStateDescriptor<>("lastVisitHomeDate", String.class));
                        lastVisitGoodDetailDateState = getRuntimeContext().getState(
                                new ValueStateDescriptor<>("lastVisitGoodDetailDate", String.class));
                    }

                    @Override
                    public void processElement(
                            JSONObject value,
                            Context ctx,
                            Collector<TrafficHomeDetailPageViewBean> out) throws Exception {
                        String pageId = value.getJSONObject("page").getString("page_id");
                        Long ts = value.getLong("ts");
                        String today = AtguiguUtil.tsToDate(ts);
                        String lastVisitHomeDate = lastVisitHomeDateState.value();
                        String lastVisitGoodDetailDate = lastVisitGoodDetailDateState.value();
                        Long homeCt = 0L;
                        Long goodDetailCt = 0L;
                        if ("home".equals(pageId)){
                            /*
                                Check whether it is the first time the user visits
                                 the home page that today.
                             */
                            if (!today.equals(lastVisitHomeDate)){
                                homeCt = 1L;
                                // upgrade the last visit home page date.
                                lastVisitHomeDateState.update(today);
                            }
                        }else {
                            /*
                                Check whether it is the first time the user visits
                                the good detail page that today.
                             */
                            if (!today.equals(lastVisitGoodDetailDate)){
                                goodDetailCt = 1L;
                                // upgrade the last visit good detail page date.
                                lastVisitGoodDetailDateState.update(today);
                            }
                        }
                        /*
                            There aer only two possibilities, the first time I go to the home page
                            or the first time I go to the good detail page, nothing else.
                         */
                        if (homeCt + goodDetailCt == 1){
                            out.collect(new TrafficHomeDetailPageViewBean(
                                    "","",
                                    homeCt,goodDetailCt,ts));
                        }

                    }
                });
    }

    private SingleOutputStreamOperator<JSONObject> filterHomeAndGoodDetail(DataStreamSource<String> ds) {
        return ds
                .map(JSON::parseObject)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((obj,ts) -> obj.getLong("ts"))
                                .withIdleness(Duration.ofSeconds(60)))
                .filter(value ->
                        "home".equals(value.getJSONObject("page").getString("page_id"))
                        || "good_detail".equals(value.getJSONObject("page").getString("page_id")));
    }
}
